﻿namespace Away.EventBusCore.Commands.Impl;

public class RemoteCommandSender : ICommandSender, ICommandHost
{
    private ConnectionMultiplexer? _conn;
    private ISubscriber? _sub;
    private readonly IServiceScope _scope;
    public RemoteCommandSender(IServiceProvider serviceProvider)
    {
        _scope = serviceProvider.CreateScope();
    }

    private IServiceProvider ServiceProvider => _scope.ServiceProvider;
    private ILogger<RemoteCommandSender> Logger => ServiceProvider.GetRequiredService<ILogger<RemoteCommandSender>>();
    private IMediator Mediator => ServiceProvider.GetRequiredService<IMediator>();
    private CommandOptionsBuilder Options => ServiceProvider.GetRequiredService<IOptionsMonitor<CommandOptionsBuilder>>().CurrentValue;

    public ConnectionMultiplexer Connection
    {
        get
        {
            if (_conn != null && _conn.IsConnecting)
            {
                return _conn;
            }
            if (string.IsNullOrWhiteSpace(Options.Host))
            {
                throw new ArgumentNullException(nameof(Options.Host), "RemoteCommandSender CommandSettings:Host is null");
            }

            _conn = ConnectionMultiplexer.Connect(new ConfigurationOptions
            {
                EndPoints = new EndPointCollection { Options.Host },
                User = Options.User,
                Password = Options.Password,
            }, new CommandLogger(Logger));
            return _conn;
        }
    }

    public ISubscriber Subscriber
    {
        get
        {
            if (_sub != null && _sub.IsConnected())
            {
                return _sub;
            }
            _sub = Connection.GetSubscriber();
            return _sub;
        }
    }

    public Task Send<TEvent>(TEvent e) where TEvent : Command
    {
        return Send(e.EventId, e);
    }

    public async Task<TResult> Send<TResult>(Command<TResult> e)
    {
        e.TraceId = $"{e.EventId}-{DateTime.Now.ToFileTime()}";
        await Send(e.EventId, e);

        // 等待返回结果
        var channel = CreateChannel(e.TraceId);
        var channelQueue = Subscriber.Subscribe(channel);
        var data = await channelQueue.ReadAsync().AsTask().WaitAsync(TimeSpan.FromMilliseconds(e.Timeout));
        Subscriber.Unsubscribe(channel);
        return JsonSerializer.Deserialize<TResult>(data.Message.ToString())!;
    }

    public Task Send(string eventId, object result)
    {
        var payload = JsonSerializer.Serialize(result);
        Logger.LogTrace("\r\n\r\npublish\r\nchannel:{}\r\npayload:{}\r\n\r\n", eventId, payload);
        return Subscriber.PublishAsync(CreateChannel(eventId), payload);
    }

    public Task Listen()
    {
        Logger.LogInformation("Command 远程模式");
        foreach (var (eventId, type) in Options.EventBusTypes)
        {
            RedisChannel channel = new(eventId, RedisChannel.PatternMode.Auto);
            Subscriber.SubscribeAsync(channel, (c, v) =>
            {
                Logger.LogTrace("\r\n\r\nlisten\r\nchannel:{}\r\npayload:{}\r\n\r\n", channel, v);
                var e = JsonSerializer.Deserialize(v.ToString(), type);
                Mediator.Send(e!);
            });
        }
        return Task.CompletedTask;
    }

    public void Dispose()
    {
        _conn?.Dispose();
        _scope.Dispose();
    }

    private static RedisChannel CreateChannel(string eventId)
    {
        return new(eventId, RedisChannel.PatternMode.Auto);
    }
}

public class CommandLogger : TextWriter
{
    private readonly ILogger _logger;
    public override Encoding Encoding => Encoding.UTF8;
    public CommandLogger(ILogger logger)
    {
        _logger = logger;
    }

    public override void WriteLine([StringSyntax("CompositeFormat")] string format, params object?[] arg)
    {
        _logger.LogInformation(format, arg);
    }
}